package com.kwin.rpcclient.rpc;

import java.net.SocketAddress;
import java.util.concurrent.SynchronousQueue;

import javax.annotation.PreDestroy;

import com.kwin.rpcclient.rpc.codec.JSONDecoder;
import com.kwin.rpcclient.rpc.codec.JSONEncoder;
import com.kwin.rpcclient.rpc.handler.NettyClientHandler;
import com.kwin.rpccommon.dto.Request;
import com.kwin.rpccommon.dto.Response;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSONArray;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.extern.slf4j.Slf4j;

/**
 *
 * @author wangkaia
 */
@Slf4j
public class NettyClient {
    
    private EventLoopGroup group = new NioEventLoopGroup(1);
    private Bootstrap bootstrap = new Bootstrap();
    
    NettyClientHandler clientHandler;
    
//    @Autowired
//    ConnectManage connectManage;
    
    
    public NettyClient(){
        clientHandler = new NettyClientHandler();
        bootstrap.group(group).
            channel(NioSocketChannel.class).
            option(ChannelOption.TCP_NODELAY, true).
            option(ChannelOption.SO_KEEPALIVE,true).
            handler(new ChannelInitializer<SocketChannel>() {
                //创建NIOSocketChannel成功后，在进行初始化时，将它的ChannelHandler设置到ChannelPipeline中，用于处理网络IO事件
                @Override
                protected void initChannel(SocketChannel channel) throws Exception {
                    ChannelPipeline pipeline = channel.pipeline();
                    pipeline.addLast(new IdleStateHandler(0, 0, 30));
                    pipeline.addLast(new JSONEncoder());
                    pipeline.addLast(new JSONDecoder());
                    pipeline.addLast("handler",clientHandler);
                }
            });
    }
    
    @PreDestroy
    public void destroy(){
        log.info("RPC客户端退出,释放资源!");
        group.shutdownGracefully();
    }
    
    public Object send(Channel channel, Request request) throws InterruptedException{
        
        if (channel!=null && channel.isActive()) {
            SynchronousQueue<Object> queue = clientHandler.sendRequest(request,channel);
            Object result = queue.take();
            return JSONArray.toJSONString(result);
        }else{
            Response res = new Response();
            res.setCode("1");
            res.setMsg("未正确连接到服务器.请检查相关配置信息!");
            return JSONArray.toJSONString(res);
        }
    }
    public Channel doConnect(SocketAddress address) throws InterruptedException {
        ChannelFuture future = bootstrap.connect(address);
        Channel channel = future.sync().channel();
        return channel;
    }
}